package com.cn.client.impl;

import com.cn.client.*;
import com.cn.client.consumer.RemoteClientOffsetStore;
import com.cn.client.remplance.AllocateMessageQueueStrategyAverage;
import com.cn.common.*;
import com.cn.util.FileConfigPropertyUtil;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DefaultMQPushConsumerImpl {

    private MQClientInstance mqClientInstance;

    private FileConfigPropertyUtil config;

    /**
     * 业务消息处理接口
     */
    private IMessageConsumeService messageConsumeService;

    /**
     * 消费端
     */
    private ConsumeMessageService consumeMessageService;

    /**
     * 消费端位移
     */
    private OffsetStore offsetStore;

    private String consumeGroup;

    /**
     * nameSrv地址
     */
    private String nameSrvAddress;

    /**
     * 下次请求间隔时间
     */
    private long repullInteral = 3000;

    private RemplanceService remplanceService;
    private RemplanceImpl remplanceImpl;
    private Set<String> topicTables = new HashSet<>();

    private AllocateMessageQueueStrategy allocateMessageQueueStrategy;

    private ScheduledExecutorService scheduledExecutorService;

    public DefaultMQPushConsumerImpl() {
        mqClientInstance = new MQClientInstance(this);
        //消息处理
        consumeMessageService = new ConsumeMessageService(mqClientInstance, this);
        //处理消费位移
        this.offsetStore = new RemoteClientOffsetStore(this.mqClientInstance);
        this.remplanceService = new RemplanceService(this);
        this.remplanceImpl = new RemplanceImpl(this, mqClientInstance);

        //负载均衡策略
        this.allocateMessageQueueStrategy = new AllocateMessageQueueStrategyAverage();
        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    }

    public void start() {
        try {
            if (StringUtils.isEmpty(this.getNameSrvAddress())) {
                config = new FileConfigPropertyUtil();
                GlobalConfigUtil.setNameSrv(config.get("namesrv_address"));
            } else {
                GlobalConfigUtil.setNameSrv(this.getNameSrvAddress());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        //注册该节点所在分组
        this.mqClientInstance.register(consumeGroup, this);
        //启动业务处理线程
        this.consumeMessageService.start();
        //负载均衡服务
        this.remplanceService.start();

        //启动
        this.mqClientInstance.start();

        //到服务端更新topic路由信息
        updateTopicChanged();
        //在更新的topic路由信息中构造pullrequest
        doRemplanceImmediately();
        System.out.println("消费节点：" + this.mqClientInstance.getClientId());

    }

    public void shutdown() {

    }

    public void dispatchPullRequest(List<PullRequest> pullRequestList) {
//        System.out.print(pullRequestList.size() + "长度开始：");
        for (PullRequest pullRequest : pullRequestList) {
            this.mqClientInstance.getPullMessageService().executePullRequestImmediately(pullRequest);
        }
    }

    public void updateTopicChanged() {
        {
            for (String topics : topicTables) {
                this.mqClientInstance.updateTopic(topics);
            }
        }
    }

    public void doRemplanceImmediately() {
        this.scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                remplanceService.wakeup();
            }
        }, 8, TimeUnit.SECONDS);//延迟执行
    }

    public void doRemplance() {
        this.remplanceImpl.doRemplance();
    }

    public void pullMsg(PullRequest pullRequest) {
        ProcessQueue processQueue = pullRequest.getProcessQueue();
        processQueue.setLastPullTimestamp(System.currentTimeMillis());
        try {
            if (!processQueue.isLocked()) {
                System.out.println("队列没有locked，定时任务锁定后可以拉取，稍后执行！");
                mqClientInstance.getPullMessageService().executePullMessageLate(pullRequest, repullInteral);
                return;
            }
            if (!processQueue.isLockFirst()) {
                long offset = this.getOffsetStore().readOffset(pullRequest.getMessageQueue());
                if (processQueue.getNextOffset() > offset) {
                    processQueue.setNextOffset(offset);
                }
                processQueue.setLockFirst(true);
            }
            if (processQueue.isDropped()) {
                System.out.println("进程已终止，稍后继续！");
                return;
            }
            if (processQueue.isPullExpired()) {
                System.out.println("拉取超时，稍后继续！");
                mqClientInstance.getPullMessageService().executePullMessageLate(pullRequest, repullInteral);
                return;
            }

            PullTask pullTask = new PullTask() {
                @Override
                public void onSuccess(PullResult result) {
                    if (result.getStatus() == PullResultStatus.FOUND) {
                        //设置下次拉取的位置
                        pullRequest.setNextOffset(result.getNextOffset());
                        pullRequest.getProcessQueue().putMessage(MessageCheckout.decodeMessage(result));
                        consumeMessageService.submit(pullRequest);
                        //准备下次拉取
                        mqClientInstance.getPullMessageService().executePullMessageLate(pullRequest, repullInteral);
                    } else {
                        //没有新的消息，立马再次拉取
                        mqClientInstance.getPullMessageService().executePullMessageImmediately(pullRequest);
                    }
                }
            };
            try {
                this.mqClientInstance.getMqClientApi().pullMessage(pullRequest, pullTask);
            } catch (Exception e) {
                mqClientInstance.getPullMessageService().executePullMessageLate(pullRequest, repullInteral);
            }
        } catch (Exception e) {
//            e.printStackTrace();
            mqClientInstance.getPullMessageService().executePullMessageLate(pullRequest, repullInteral);
        }
    }

    public void reshMessageQueue(String topic, Set<MessageQueue> messageQueueSet) {
        this.remplanceImpl.messageQueueTables.put(topic, messageQueueSet);
    }

    public void setMessageConsumeService(IMessageConsumeService messageConsumeService) {
        this.messageConsumeService = messageConsumeService;
    }

    public OffsetStore getOffsetStore() {
        return offsetStore;
    }

    public IMessageConsumeService getMessageConsumeService() {
        return messageConsumeService;
    }

    public String getNameSrvAddress() {
        return nameSrvAddress;
    }

    public void setNameSrvAddress(String nameSrvAddress) {
        this.nameSrvAddress = nameSrvAddress;
    }


    public RemplanceService getRemplanceService() {
        return remplanceService;
    }

    public RemplanceImpl getRemplanceImpl() {
        return remplanceImpl;
    }

    public String getConsumeGroup() {
        return consumeGroup = StringUtils.isEmpty(consumeGroup) ? "default" : consumeGroup;
    }

    public Set<String> getTopicTables() {
        return topicTables;
    }

    public void setTopicTables(Set<String> topicTables) {
        this.topicTables = topicTables;
    }

    public void setConsumeGroup(String consumeGroup) {
        this.consumeGroup = consumeGroup;
    }

    public long getRepullInteral() {
        return repullInteral;
    }

    public void setRepullInteral(long repullInteral) {
        this.repullInteral = repullInteral;
    }

    public AllocateMessageQueueStrategy getAllocateMessageQueueStrategy() {
        return allocateMessageQueueStrategy;
    }

    public static void main(String[] args) {
        IMessageConsumeService messageConsumeService = new IMessageConsumeService() {
            @Override
            public MessageConsumeResultStatus execute(List<MessageExt> msgs) {
                for (MessageExt msg : msgs) {
                    System.out.println("消息内容：" + new String(msg.getBody(), Charset.forName("utf-8")));
                }
                return MessageConsumeResultStatus.commit;
            }
        };
        DefaultMQPushConsumerImpl defaultMQPushConsumer = new DefaultMQPushConsumerImpl();
        defaultMQPushConsumer.getTopicTables().add("topic");
//        defaultMQPushConsumer.getTopicTables().add("demo");
        defaultMQPushConsumer.setConsumeGroup("yaojin_demo");
        defaultMQPushConsumer.setNameSrvAddress("127.0.0.1:2020");
        defaultMQPushConsumer.setMessageConsumeService(messageConsumeService);

        GlobalConfigUtil.setNameSrv("127.0.0.1:2020");
        defaultMQPushConsumer.start();
    }

}
